package drds.binlog.binlog_task;

import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.Authors;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.common.utils.JsonUtils;
import drds.binlog.data_object.Entry;
import drds.binlog.event_filter.aviater.RegexFilter;
import drds.binlog.metadata.IMetaDataManager;
import drds.binlog.metadata.MetaDataManager;
import drds.binlog.parse.AbstractEventParser;
import drds.binlog.parse.GroupEventParser;
import drds.binlog.parse.IEventParser;
import drds.binlog.parse.log_position_manager.LogPositionManager;
import drds.binlog.parse.log_position_manager.MemoryLogPositionManager;
import drds.binlog.parse.mysql.AuthenticationInfo;
import drds.binlog.parse.mysql.EventParser;
import drds.binlog.sink.AbstractEntrySink;
import drds.binlog.sink.EntrySink;
import drds.binlog.sink.IEntrySink;
import drds.binlog.sink.group.GroupEntrySink;
import drds.binlog.store.AbstractCleanUp;
import drds.binlog.store.Event;
import drds.binlog.store.EventStore;
import drds.common.$;
import drds.common.Author;
import drds.common.Constants;
import drds.common.Else;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;


@Slf4j
@Author(name = Authors.LI_YANG)
public class BinlogTask extends AbstractLifeCycle implements IBinlogTask
{


    @Setter
    @Getter
    protected Long binlogTaskId;
    @Setter
    @Getter
    protected Parameter parameter;                                                      // 对应参数

    @Setter
    @Getter
    protected String destination;                                                  // 队列名字


    //对个实例进行原数据进行记录
    @Setter
    @Getter
    protected IMetaDataManager metaDataManager;                                                  // 消费信息管理器

    //对个实例进行解析
    @Setter
    @Getter
    protected IEventParser eventParser;                                                  // 解析对应的数据信息
    @Setter
    @Getter
    protected String schemaNameAndTableNameRegexFilter;                                                          // 过滤表达式
    @Setter
    @Getter
    protected IEntrySink<List<Entry>> entrySink;                                                    // 链接parse和store的桥接器
    @Setter
    @Getter
    protected drds.binlog.store.IEventStore<Event> eventStore;                                                   // 有序队列


    public BinlogTask(BinlogTaskInfo binlogTaskInfo)
    {

        this.binlogTaskId = binlogTaskInfo.getBinlogTaskId();
        this.parameter = binlogTaskInfo.getParameter();
        //
        this.destination = binlogTaskInfo.getDestination();
        //
        this.schemaNameAndTableNameRegexFilter = parameter.getSchemaNameAndTableNameRegexFilter();
        log.info(Constants.right);
        log.info("init binlog消费服务启动[{}] begin", this.binlogTaskId);
        // 初始化metaManager
        initMetaManager();
        // 初始化eventStore
        initEventStore();
        // 初始化eventSink
        initEventSink();
        // 初始化eventParser;
        initEventParser();


        if (!metaDataManager.isRunning())
        {
            metaDataManager.start();
        }
        log.info("init binlog消费服务启动[{}] end", this.binlogTaskId);
        log.info(Constants.left);
    }

    @Override
    public boolean subscribeChange(ClientInfo clientInfo)
    {
        if ($.isNotNullAndNotEmpty(clientInfo.getFilter()))
        {
            RegexFilter regexFilter = new RegexFilter(clientInfo.getFilter());
            boolean isGroupEventParser = (eventParser instanceof GroupEventParser);
            if (isGroupEventParser)
            {
                List<IEventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
                for (IEventParser eventParser : eventParserList)
                {
                    ((AbstractEventParser) eventParser).setSchemaNameAndTableNameRegexFilter(regexFilter);
                }
            } else
            {
                ((AbstractEventParser) eventParser).setSchemaNameAndTableNameRegexFilter(regexFilter);
            }

        }

        return true;
    }

    @Override
    public void start()
    {

        log.info("启动" + BinlogTask.class.getSimpleName() + " for {}-{} with parameter:{}", binlogTaskId, destination, parameter);

        super.start();
        //
        if (!metaDataManager.isRunning())
        {
            metaDataManager.start();
        }
        //后面三个顺序不能颠倒
        //
        if (!eventStore.isRunning())
        {
            eventStore.start();
        }
        //
        if (!entrySink.isRunning())
        {
            entrySink.start();
        }
        //
        if (!eventParser.isRunning())
        {
            beforeStartEventParser(eventParser);
            eventParser.start();
            afterStartEventParser(eventParser);
        }
        log.info("start successful....");
    }

    @Override
    public void stop()
    {
        super.stop();
        log.info("stop  for {}-{} ", new Object[]{binlogTaskId, destination});
        //后面三个顺序不能颠倒
        if (eventParser.isRunning())
        {
            beforeStopEventParser(eventParser);
            eventParser.stop();
            afterStopEventParser(eventParser);
        }

        if (entrySink.isRunning())
        {
            entrySink.stop();
        }

        if (eventStore.isRunning())
        {
            eventStore.stop();
        }
        //
        if (metaDataManager.isRunning())
        {
            metaDataManager.stop();
        }


        log.info("stop successful....");
    }

    protected void beforeStartEventParser(IEventParser eventParser)
    {

        boolean isGroupEventParser = (eventParser instanceof GroupEventParser);
        if (isGroupEventParser)
        {
            List<IEventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
            for (IEventParser $eventParser : eventParserList)
            {
                startEventParserInternal($eventParser);
            }
        } else
        {
            startEventParserInternal(eventParser);
        }
    }

    // around event parser, default impl
    protected void afterStartEventParser(IEventParser eventParser)
    {
        log.debug("eventParser:" + eventParser);
        // 读取一下历史订阅的filter信息
        List<ClientInfo> clientInfoList = metaDataManager.getClientInfoList(destination);
        for (ClientInfo clientInfo : clientInfoList)
        {
            subscribeChange(clientInfo);
        }
    }

    // around event parser
    protected void beforeStopEventParser(IEventParser eventParser)
    {
        log.debug("eventParser:" + eventParser);
    }

    protected void afterStopEventParser(IEventParser eventParser)
    {

        boolean isGroupEventParser = (eventParser instanceof GroupEventParser);
        if (isGroupEventParser)
        {
            List<IEventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
            for (IEventParser $eventParser : eventParserList)
            {// 需要遍历启动
                stopEventParserInternal($eventParser);
            }
        } else
        {
            stopEventParserInternal(eventParser);
        }
    }

    /**
     * 初始化单个eventParser，不需要考虑group
     */
    protected void startEventParserInternal(IEventParser eventParser)
    {
        if (eventParser instanceof AbstractEventParser)
        {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            LogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (!logPositionManager.isRunning())
            {
                logPositionManager.start();
            }
        }


    }

    protected void stopEventParserInternal(IEventParser eventParser)
    {
        if (eventParser instanceof AbstractEventParser)
        {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            LogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (logPositionManager.isRunning())
            {
                logPositionManager.stop();
            }
        }


    }


    @Override
    public drds.binlog.store.IEventStore getEventStore()
    {
        return eventStore;
    }

    @Override
    public IEntrySink getEventSink()
    {
        return entrySink;
    }

    @Override
    public IEventParser getEventParser()
    {
        return eventParser;
    }

    @Override
    public IMetaDataManager getMetaDataManager()
    {
        return metaDataManager;
    }

    @Override
    public String getDestination()
    {
        return destination;
    }


    protected void initMetaManager()
    {

        log.info("init metaDataManager[{}] begin", this.binlogTaskId);
        metaDataManager = new MetaDataManager();
        log.info("init metaDataManager[{}] end", this.binlogTaskId);

    }

    protected void initEventStore()
    {

        log.info("init eventStore[{}] begin", this.binlogTaskId);
        EventStore eventStore = new EventStore();
        eventStore.setEventsSize(parameter.getMemoryStorageBufferSize());
        eventStore.setMemorySizePerEvent(parameter.getMemoryStorageBufferMemUnit());
        eventStore.setStorageBatchMode(parameter.getStorageBatchMode());
        eventStore.setDdlIsolation(parameter.getDdlIsolation());
        this.eventStore = eventStore;

        if (this.eventStore instanceof AbstractCleanUp)
        {
            StorageScavengeMode storageScavengeMode = parameter.getStorageScavengeMode();
            AbstractCleanUp abstractCleanUp = (AbstractCleanUp) this.eventStore;
            abstractCleanUp.setDestination(destination);
            abstractCleanUp.setMetaDataManager(metaDataManager);
            abstractCleanUp.setOnAck(storageScavengeMode.isOnAck());
            abstractCleanUp.setOnFull(storageScavengeMode.isOnFull());
            abstractCleanUp.setOnSchedule(storageScavengeMode.isOnSchedule());
            if (storageScavengeMode.isOnSchedule())
            {
                abstractCleanUp.setScavengeSchedule(parameter.getScavengeSchdule());
            }
        }
        log.info("init eventStore[{}] end", this.binlogTaskId);
    }

    protected void initEventSink()
    {
        log.info("init eventSink[{}] begin", this.binlogTaskId);

        int salveInfoListSize = SalveInfoListSize();
        if (salveInfoListSize <= 1)
        {
            entrySink = new EntrySink();
        } else
        {
            entrySink = new GroupEntrySink(salveInfoListSize);
        }
        if (entrySink instanceof EntrySink)
        {
            ((EntrySink) entrySink).setFilterTransactionEvent(false);
            ((EntrySink) entrySink).setEventStore(this.getEventStore());
        }
        if (StringUtils.isNotEmpty(this.schemaNameAndTableNameRegexFilter))
        {
            RegexFilter regexFilter = new RegexFilter(this.schemaNameAndTableNameRegexFilter);
            ((AbstractEntrySink<List<Entry>>) entrySink).setEventFilter(regexFilter);
        }

        log.info("init eventSink[{}] end", this.binlogTaskId);
    }

    protected void initEventParser()
    {

        log.info("init eventParser[{}] begin", this.binlogTaskId);
        if (!$.isHasData(parameter.salveInfoList))
        {
            throw new Else();
        }

        List<IEventParser> eventParserList = new ArrayList<IEventParser>();
        for (SalveInfo salveInfo : parameter.salveInfoList)
        {
            eventParserList.add(doInitEventParser(salveInfo));
        }


        if (eventParserList.size() > 1)
        { // 如果存在分组，构造分组的parser
            GroupEventParser groupEventParser = new GroupEventParser();
            groupEventParser.setEventParserList(eventParserList);
            this.eventParser = groupEventParser;
        } else
        {
            this.eventParser = eventParserList.get(0);
        }

        log.info("init eventParser[{}] end", this.binlogTaskId);
    }

    private IEventParser doInitEventParser(SalveInfo salveInfo)
    {

        EventParser eventParser = new EventParser();
        eventParser.setDestination(destination);
        // 编码参数
        eventParser.setConnectionCharset(Charset.forName(parameter.getConnectionCharset()));
        eventParser.setConnectionCharsetNumber(parameter.getConnectionCharsetNumber());
        // 网络相关参数
        eventParser.setDefaultConnectionTimeoutInSeconds(parameter.getDefaultConnectionTimeoutInSeconds());
        eventParser.setSendBufferSize(parameter.getSendBufferSize());
        eventParser.setReceiveBufferSize(parameter.getReceiveBufferSize());
        // 心跳检查参数
        eventParser.setDetectingEnable(parameter.getDetectingEnable());
        eventParser.setDetectingSQL(parameter.getDetectingSQL());
        eventParser.setDetectingIntervalInSeconds(parameter.getDetectingIntervalInSeconds());
        // 数据库信息参数
        AuthenticationInfo authenticationInfo = new AuthenticationInfo(new InetSocketAddress(salveInfo.getHost(), salveInfo.getPort()), salveInfo.getDatabaseName(), salveInfo.getUsername(), salveInfo.getPassword());
        eventParser.setAuthenticationInfo(authenticationInfo);

        if ($.isHasData(parameter.getPositions()))
        {
            EntryPosition position = JsonUtils.unmarshalFromString(parameter.getPositions().get(0),
                    EntryPosition.class);
            // binlog位置参数
            eventParser.setEntryPosition(position);


        }
        eventParser.setFallbackIntervalInSeconds(parameter.getFallbackIntervalInSeconds());
        eventParser.setFilterTableError(parameter.getFilterTableError());
        this.eventParser = eventParser;


        if (this.eventParser instanceof AbstractEventParser)
        {
            AbstractEventParser abstractEventParser = (AbstractEventParser) this.eventParser;
            abstractEventParser.setTransactionSize(parameter.getTransactionSize());
            abstractEventParser.setLogPositionManager(initLogPositionManager());

            abstractEventParser.setEntrySink(getEventSink());


            if ($.isNotNullAndNotEmpty(schemaNameAndTableNameRegexFilter))
            {
                RegexFilter schemaNameAndTableNameRegexFilter = new RegexFilter(this.schemaNameAndTableNameRegexFilter);
                abstractEventParser.setSchemaNameAndTableNameRegexFilter(schemaNameAndTableNameRegexFilter);
            }

            // 设置黑名单
            if ($.isNotNullAndNotEmpty(parameter.getBlackSchemaNameAndTableNameRegexFilter()))
            {
                RegexFilter blackSchemaNameAndTableNameRegexFilter = new RegexFilter(parameter.getBlackSchemaNameAndTableNameRegexFilter());
                abstractEventParser.setBlackSchemaNameAndTableNameRegexFilter(blackSchemaNameAndTableNameRegexFilter);
            }
        }

        return this.eventParser;
    }


    protected LogPositionManager initLogPositionManager()
    {


        log.info("init logPositionPersistManager[{}] begin", this.binlogTaskId);
        LogPositionManager logPositionManager = new MemoryLogPositionManager();
        log.info("init logPositionPersistManager[{}] end", this.binlogTaskId);

        return logPositionManager;
    }


    private int SalveInfoListSize()
    {
        return parameter.getSalveInfoList().size();
    }


}
